package cn.doitedu.demo3;

import cn.doitedu.demo1.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/11/18
 * @Desc: 学大数据，上多易教育
 *
 *
 * -- 营销策略1：营销规则
 *    针对月收入在 20~28 之间，价值等级 < 2的用户，                     -- 静态画像条件
 *    当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息   -- 触发条件
 *
 * -- 营销策略1：营销规则
 *    针对月收入在 20~28 之间，价值等级 < 2的用户，                     -- 静态画像条件
 *    规则上线后，用户要做过收藏行为 3+ 次
 *    当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息   -- 触发条件
 *
 *
 **/
public class RuleModel1_Calculator implements RuleCalculatorV3 {

    RestHighLevelClient client;
    SearchRequest request;
    JSONObject message;
    JSONObject paramJsonObject;

    @Override
    public void init(RuntimeContext runtimeContext,String ruleParam) {

        paramJsonObject = JSON.parseObject(ruleParam);
        String ruleId = paramJsonObject.getString("rule_id");


        // es的请求客户端
        client = new RestHighLevelClient(
                RestClient
                        .builder(new HttpHost("doitedu", 9200, "http"))
        );
        // 用于查询参数封装的对象
        request = new SearchRequest("doit42_userprofile");


        message = new JSONObject();
        message.put("rule_id", ruleId);
    }

    @Override
    public void calculate(EventBean eventBean, Collector<String> out) throws IOException {
        // 判断当前事件，是否是规则的触发事件
        String triggerEvent = paramJsonObject.getString("trigger_event");
        if (triggerEvent.equals(eventBean.getEvent_id())) {

            // 判断当前行为人，是否满足营销规则中的画像条件
            IdsQueryBuilder idQuery = QueryBuilders.idsQuery().addIds(eventBean.getUser_id() + "");

            // 从参数中取第1个静态画像条件
            JSONArray staticProfileArray = paramJsonObject.getJSONArray("static_profile");
            JSONObject staticProfile_0 = staticProfileArray.getJSONObject(0);
            String tagName0 = staticProfile_0.getString("tag_name");
            JSONArray valueRange0 = staticProfile_0.getJSONArray("value_range");
            int value0_0 = valueRange0.getIntValue(0);
            int value0_1 = valueRange0.getIntValue(1);

            /*拼装第1个静态画像条件的es查询条件*/
            RangeQueryBuilder ageQuery = QueryBuilders.rangeQuery(tagName0).gte(value0_0).lte(value0_1);

            /*拼装第2个静态画像条件的es查询条件*/
            JSONObject staticProfile_1 = staticProfileArray.getJSONObject(1);
            String tagName1 = staticProfile_1.getString("tag_name");
            JSONArray valueRange1 = staticProfile_1.getJSONArray("value_range");
            String value1_0 = valueRange1.getString(0);
            MatchQueryBuilder genderQuery = QueryBuilders.matchQuery(tagName1, value1_0);

            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            boolQueryBuilder.must(idQuery)
                    .must(ageQuery)
                    .must(genderQuery);


            request.source(new SearchSourceBuilder().query(boolQueryBuilder));

            // 用客户端发送请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            if (response.getHits().getHits().length > 0) {
                // 如果满足，则发送消息
                message.put("user_id", eventBean.getUser_id());
                message.put("hit_time", eventBean.getAction_time());
                out.collect(message.toJSONString());
            }
        }
    }
}
